using System;
using System.Text;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Microsoft.Extensions.Configuration;
using NewLife.RocketMQ;
using System.Threading.Tasks;
using System.Collections.Generic;
using ZXing;
using System.Runtime.InteropServices;

namespace QQZiFramework.MQ
{
    public class RocketMQ : IMQDriver
    {
        string hostName, port, userName, password;

        IConfiguration Configuration;


        public RocketMQ(IConfiguration configuration)
        {
            Configuration = configuration;
            var configItem = configuration.GetSection("RocketMQ");
            hostName = configItem["HostName"];
            port = configItem["Port"];
            // userName = configItem["UserName"];
            // password = configItem["Password"];
            // if (configItem["AutoAck"] != null)
            //     AutoAck = Convert.ToBoolean(configItem["AutoAck"]);
            if (string.IsNullOrEmpty(hostName)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:HostName");
            if (string.IsNullOrEmpty(port)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:Port");
            // if (string.IsNullOrEmpty(userName)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:UserName");
            // if (string.IsNullOrEmpty(password)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:Password");
            // var queues = configItem.GetSection("Queues");
            // if (queues != null)
            // {

            //     if (queues["BasicQos"] != null) BasicQos = Convert.ToUInt16(queues["BasicQos"]);
            //     if (queues["Durable"] != null) Durable = Convert.ToBoolean(queues["Durable"]);

            // }

        }
        System.Collections.Generic.Dictionary<string, Producer> channels = new System.Collections.Generic.Dictionary<string, Producer>();

        public void Publish(string queueName, object obj)
        {
            Producer mq;

            if (!channels.TryGetValue(queueName, out mq))
            {
                mq = new Producer
                {
                    NameServerAddress = hostName + ":" + port
                    //,Log = XTrace.Log,
                };
                mq.Start();
                mq.CreateTopic(queueName, 4);
                mq.Dispose();
                mq = new Producer
                {
                    NameServerAddress = hostName + ":" + port,
                    Topic = queueName
                    //,Log = XTrace.Log,
                };
                mq.Start();
                channels.Add(queueName, mq);
            }

            string body;
            Type type = obj.GetType();
            if (type.IsValueType || type == typeof(string))
            {
                body = obj.ToString();
            }
            else if (type == typeof(byte[]))
            {
                throw new Exception("rocketmq客户端暂不支持byte[]");
            }
            else
            {
                // string itemJson;
                if (obj is IDBEntity) body = (obj as IDBEntity).ToJson();
                else body = JsonConvert.SerializeObject(obj);
            }
            mq.Publish(body);
        }

        public long GetLength(string queueName)
        {
            Producer item;
            if (!channels.TryGetValue(queueName, out item))
            {
                throw new Exception("仅发布者所在程序可以获取！");
            }
            else
            {
                return -1;
                // QueueDeclareOk result = item.Channel.QueueDeclare();
                // return (long)result.MessageCount;
            }

        }

        public void BindReceived(string queueName, Func<byte[], bool> fun)
        {
            var consumer = new Consumer
            {
                Topic = queueName,
                NameServerAddress = hostName + ":" + port,
                FromLastOffset = true,
                SkipOverStoredMsgCount = 0,
                BatchSize = 20
                //,Log = XTrace.Log,
            };
            consumer.OnConsume = (q, ms) =>
            {
                //XTrace.WriteLine("[{0}@{1}]收到消息[{2}]", q.BrokerName, q.QueueId, ms.Length);
                foreach (var item in ms)
                {
                    fun.Invoke(item.Body);
                    //XTrace.WriteLine($"消息：主键【{item.Keys}】，产生时间【{item.BornTimestamp.ToDateTime()}】，内容【{item.Body.ToStr()}】");
                }
                return true;
            };
            consumer.Start();
        }

        public void BindReceivedAsync(string queueName, Func<byte[], Task<bool>> fun)
        {
            var consumer = new Consumer
            {
                Topic = queueName,
                NameServerAddress = hostName + ":" + port,
                FromLastOffset = true,
                SkipOverStoredMsgCount = 0,
                BatchSize = 20
                //,Log = XTrace.Log,
            };
            consumer.OnConsume = (q, ms) =>
            {
                //XTrace.WriteLine("[{0}@{1}]收到消息[{2}]", q.BrokerName, q.QueueId, ms.Length);
                foreach (var item in ms)
                {
                    fun.Invoke(item.Body);
                    //XTrace.WriteLine($"消息：主键【{item.Keys}】，产生时间【{item.BornTimestamp.ToDateTime()}】，内容【{item.Body.ToStr()}】");
                }
                return true;
            };
            consumer.Start();
        }
    }
}
